package org.zn.etl.jdbc;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.function.Consumer;

import org.apache.commons.lang3.tuple.Triple;
import org.zn.datax.common.entity.ColumnInfo;
import org.zn.etl.job.handler.StateOfRWBuffer;

import com.alibaba.druid.util.JdbcUtils;
import com.google.common.collect.Maps;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ResultSetTools {

	public static StateOfRWBuffer<Map<String, Object>> readToStateOfRWBuffer(ResultSet rs,Connection conn,Statement stat, Integer size)
			throws SQLException {

		StateOfRWBuffer<Map<String, Object>> buff = new StateOfRWBuffer<>(size);

		Map<String, ColumnInfo> colMapping = columnMetaData(rs);

		Thread thread = new Thread(() -> {
			try {
				int rows=0;
				int max=0;
				
				while (rs.next()) {

					buff.doWrite();

					Map<String, Object> m = Maps.newHashMap();
					for (Entry<String, ColumnInfo> col : colMapping.entrySet()) {

						String type = col.getValue().getType();
						Integer jdbcType = col.getValue().getJdbcType();
						String key = col.getKey();

						Object value = getValue(rs, key, type); // getValue(rs, key, type.toLowerCase());

						m.put(key, value);
					}
					buff.add(m);
					
					rows++;
					max++;
					
					if(rows==size) {
						buff.ready();
						rows=0;
						buff.waitToWrite();
					}
					
				}
				
				buff.ready();
				buff.waitToWrite();
				
				log.info("ResultSet read max:{}",max);
				

			} catch (SQLException | InterruptedException e) {
				log.error("", e);
			}
			finally {

				
				buff.close();
				
				JdbcUtils.close(rs);
				JdbcUtils.close(stat);
				JdbcUtils.close(conn);
			
			}
		});

		thread.start();

		return buff;

	}

	public static List<Map<String, Object>> toMapReslt(ResultSet rs) throws SQLException {

		List<Map<String, Object>> resList = new ArrayList<>();

		Map<String, ColumnInfo> colMapping = columnMetaData(rs);

		int rows = 0;
		while (rs.next()) {

			Map<String, Object> m = Maps.newHashMap();

			for (Entry<String, ColumnInfo> col : colMapping.entrySet()) {

				String type = col.getValue().getType();
				Integer jdbcType = col.getValue().getJdbcType();
				String key = col.getKey();

				Object value = getValue(rs, key, type); // getValue(rs, key, type.toLowerCase());

				m.put(key, value);
			}
			resList.add(m);
		}

		return resList;

	}

	public static Map<String, ColumnInfo> columnMetaData(ResultSet rs) throws SQLException {
		ResultSetMetaData rsmed = rs.getMetaData();
		int ccount = rsmed.getColumnCount();
		Map<String, ColumnInfo> cim = new LinkedHashMap<String, ColumnInfo>();

		for (int i = 1; i < ccount + 1; i++) {
			String name = rsmed.getColumnLabel(i);
			String type = rsmed.getColumnTypeName(i);
			Integer jdbcType = rsmed.getColumnType(i);

			int precision = rsmed.getPrecision(i);

			ColumnInfo ci = new ColumnInfo();
			ci.setName(name.toLowerCase());
			ci.setType(type);
			ci.setLength(precision);
			ci.setJdbcType(jdbcType);

			cim.put(name.toLowerCase(), ci);

		}

		return cim;

	}

	private static Object getValue(ResultSet rs, String key) throws SQLException {

		return rs.getObject(key);

	}

	private static Object getValue(ResultSet rs, String key, String dtype) throws SQLException {

		Object value = null;

		String type = dtype.toLowerCase();

		if (TypeSets.STRING_TYPE.contains(type)) {
			value = rs.getString(key);
		} else if (TypeSets.INT_TYPE.contains(type)) {
			value = rs.getInt(key);
		} else if (TypeSets.LONG_TYPE.contains(type)) {
			value = rs.getLong(key);
		} else if (TypeSets.DECIMA_LTYPE.contains(type)) {
			value = rs.getBigDecimal(key);
		} else if (TypeSets.DATE_TYPE.contains(type)) {
			value = rs.getDate(key);
		} else if (TypeSets.BYTES_TYPE.contains(type)) {
			value = rs.getBytes(key);
		} else if (TypeSets.DOUBLE_TYPE.contains(type)) {
			value = rs.getDouble(key);
		} else if (TypeSets.BOOLEAN_TYPE.contains(type)) {
			value = rs.getBoolean(key);
		}

		return value;

	}

}
